package org.example.api.transform;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/2/19
 */
public class TransformByRollingAggregation {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        // Custom code
//        SingleOutputStreamOperator<SensorReading> mapStream = inputSteam.map(new MapFunction<String, SensorReading>() {
//            @Override
//            public SensorReading map(String line) throws Exception {
//                String[] fields = line.split(",");
//                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
//            }
//        });

        // Lamda code
        SingleOutputStreamOperator<SensorReading> mapStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // key by
        KeyedStream<SensorReading, Tuple> keyedStream = mapStream.keyBy("id");
        // KeyedStream<SensorReading, String> keyedStream1 = mapStream.keyBy(SensorReading::getId);

        // rolling aggregation, get the maximum of temperature
        SingleOutputStreamOperator<SensorReading> resultStream = keyedStream.maxBy("temperature");

        resultStream.print();

        env.execute();
    }
}
